package com.example.yckjbigdataservice.utils;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.regionserver.DisabledRegionSplitPolicy;
import org.apache.hadoop.hbase.regionserver.KeyPrefixRegionSplitRestriction;
import org.apache.hadoop.hbase.regionserver.DelimitedKeyPrefixRegionSplitRestriction;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.conf.Configuration;

public class HBaseOptimizedTableConfiguration {

    public static void main(String[] args) throws Exception {
        Configuration config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", "localhost");
        config.set("hbase.zookeeper.property.clientPort", "2181");

        try (Connection connection = ConnectionFactory.createConnection(config);
             Admin admin = connection.getAdmin()) {

            // 示例1：时序数据表优化配置
            createTimeSeriesOptimizedTable(admin);

            // 示例2：用户数据表优化配置
            createUserDataOptimizedTable(admin);

            // 示例3：日志数据表优化配置
            createLogDataOptimizedTable(admin);

            // 示例4：OLAP分析表优化配置
            createAnalyticsOptimizedTable(admin);
        }
    }

    /**
     * 时序数据表优化配置
     * 特点：写多读少，按时间范围查询，数据压缩比高
     */
    public static void createTimeSeriesOptimizedTable(Admin admin) throws Exception {
        TableName tableName = TableName.valueOf("timeseries_metrics");

        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }

        TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(tableName);

        // === Region Split Restriction 配置 ===
        // 使用KeyPrefixRegionSplitRestriction，防止相同设备的数据被分割到不同region
        tableBuilder.setRegionSplitPolicyClassName(
                KeyPrefixRegionSplitRestriction.class.getName());
        // 设置前缀长度（假设rowkey格式：deviceId_timestamp，设备ID长度为8）
        tableBuilder.setValue(KeyPrefixRegionSplitRestriction.PREFIX_LENGTH_KEY, "8");

        // === 表级别配置 ===
        tableBuilder.setMaxFileSize(2L * 1024 * 1024 * 1024); // 2GB，时序数据单个region不宜过大
        tableBuilder.setMemStoreFlushSize(128L * 1024 * 1024); // 128MB
        tableBuilder.setCompactionEnabled(true);

        // === 时序数据列族配置 ===
        ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder
                .newBuilder("metrics".getBytes());

        // 压缩配置 - 时序数据压缩效果好
        cfBuilder.setCompressionType(Compression.Algorithm.SNAPPY); // 主压缩算法
        cfBuilder.setCompactionCompressionType(Compression.Algorithm.LZ4); // Compaction时的压缩算法

        // 数据块编码 - 时序数据重复度高，效果显著
        cfBuilder.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);

        // BloomFilter配置 - 时序数据按rowkey查询多
        cfBuilder.setBloomFilterType(BloomType.ROW);
        // 注意：BloomFilter误判率由HBase自动管理，无法手动设置

        // Block缓存配置
        cfBuilder.setBlockCacheEnabled(true);
        cfBuilder.setCacheDataOnWrite(false); // 写时不缓存，减少缓存污染
        cfBuilder.setPrefetchBlocksOnOpen(false); // 不预取，按需加载
        // 注意：索引和BloomFilter缓存、L1缓存等由HBase系统自动管理

        // 版本和TTL
        cfBuilder.setMaxVersions(1); // 时序数据通常只需要最新版本
        cfBuilder.setTimeToLive(30 * 24 * 3600); // 30天TTL
        cfBuilder.setMinVersions(0);

        // HFile配置
        cfBuilder.setBlocksize(32 * 1024); // 32KB，时序数据小块有利于压缩
        cfBuilder.setInMemory(false); // 不常驻内存

        tableBuilder.setColumnFamily(cfBuilder.build());

        // 预分割 - 基于设备ID范围
        byte[][] splitKeys = generateTimeSeriesSplitKeys();
        admin.createTable(tableBuilder.build(), splitKeys);

        System.out.println("时序数据表创建成功：" + tableName.getNameAsString());
        printTableConfiguration("时序数据表", cfBuilder.build());
    }

    /**
     * 用户数据表优化配置
     * 特点：读写并重，随机访问多，需要快速检索
     */
    public static void createUserDataOptimizedTable(Admin admin) throws Exception {
        TableName tableName = TableName.valueOf("user_profiles");

        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }

        TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(tableName);

        // === Region Split Restriction 配置 ===
        // 使用DelimitedKeyPrefixRegionSplitRestriction
        tableBuilder.setRegionSplitPolicyClassName(
                DelimitedKeyPrefixRegionSplitRestriction.class.getName());
        tableBuilder.setValue(DelimitedKeyPrefixRegionSplitRestriction.DELIMITER_KEY, "_");

        tableBuilder.setMaxFileSize(5L * 1024 * 1024 * 1024); // 5GB
        tableBuilder.setMemStoreFlushSize(256L * 1024 * 1024); // 256MB

        // === 基本信息列族 ===
        ColumnFamilyDescriptorBuilder basicCfBuilder = ColumnFamilyDescriptorBuilder
                .newBuilder("basic".getBytes());

        // 压缩配置 - 用户数据中等压缩
        basicCfBuilder.setCompressionType(Compression.Algorithm.SNAPPY);
        basicCfBuilder.setCompactionCompressionType(Compression.Algorithm.SNAPPY);

        // 数据块编码
        basicCfBuilder.setDataBlockEncoding(DataBlockEncoding.PREFIX);

        // BloomFilter - 用户数据随机访问多，ROW级别效果好
        basicCfBuilder.setBloomFilterType(BloomType.ROW);
        // 注意：BloomFilter误判率由HBase自动优化

        // Block缓存 - 用户数据热点访问
        basicCfBuilder.setBlockCacheEnabled(true);
        basicCfBuilder.setCacheDataOnWrite(true); // 写时缓存，用户数据可能立即被读取
        basicCfBuilder.setPrefetchBlocksOnOpen(true); // 预取热点数据
        // 注意：索引缓存和L1缓存由HBase系统管理

        basicCfBuilder.setMaxVersions(3); // 保留3个版本
        basicCfBuilder.setBlocksize(64 * 1024); // 64KB标准块大小
        basicCfBuilder.setInMemory(false);

        tableBuilder.setColumnFamily(basicCfBuilder.build());

        // === 扩展信息列族 ===
        ColumnFamilyDescriptorBuilder extCfBuilder = ColumnFamilyDescriptorBuilder
                .newBuilder("extended".getBytes());

        // 扩展信息访问频率较低，更注重存储效率
        extCfBuilder.setCompressionType(Compression.Algorithm.LZ4);
        extCfBuilder.setCompactionCompressionType(Compression.Algorithm.ZSTD);
        extCfBuilder.setDataBlockEncoding(DataBlockEncoding.DIFF);

        extCfBuilder.setBloomFilterType(BloomType.ROWCOL); // 列级别布隆过滤器
        // 注意：BloomFilter误判率由HBase自动管理

        extCfBuilder.setBlockCacheEnabled(true);
        extCfBuilder.setCacheDataOnWrite(false);
        // 注意：索引和L1缓存由HBase系统自动管理

        extCfBuilder.setMaxVersions(1);
        extCfBuilder.setBlocksize(128 * 1024); // 128KB，减少块数量

        tableBuilder.setColumnFamily(extCfBuilder.build());

        admin.createTable(tableBuilder.build());

        System.out.println("用户数据表创建成功：" + tableName.getNameAsString());
        printTableConfiguration("用户基本信息列族", basicCfBuilder.build());
        printTableConfiguration("用户扩展信息列族", extCfBuilder.build());
    }

    /**
     * 日志数据表优化配置
     * 特点：写多读少，顺序写入，需要高压缩比
     */
    public static void createLogDataOptimizedTable(Admin admin) throws Exception {
        TableName tableName = TableName.valueOf("application_logs");

        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }

        TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(tableName);

        // === Region Split Restriction 配置 ===
        // 禁用自动分割限制，让系统根据大小自动分割
        tableBuilder.setRegionSplitPolicyClassName(
                DisabledRegionSplitPolicy.class.getName());

        tableBuilder.setMaxFileSize(10L * 1024 * 1024 * 1024); // 10GB，日志数据可以较大
        tableBuilder.setMemStoreFlushSize(512L * 1024 * 1024); // 512MB，批量写入

        ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder
                .newBuilder("log".getBytes());

        // 最大压缩配置 - 日志数据重复内容多
        cfBuilder.setCompressionType(Compression.Algorithm.ZSTD); // 主压缩算法
        cfBuilder.setCompactionCompressionType(Compression.Algorithm.ZSTD); // Compaction压缩算法

        // 数据块编码 - 日志数据结构化程度高
        cfBuilder.setDataBlockEncoding(DataBlockEncoding.PREFIX);

        // BloomFilter - 日志数据很少随机查询，可以关闭节省空间
        cfBuilder.setBloomFilterType(BloomType.NONE);

        // Block缓存 - 日志数据写多读少，最小化缓存使用
        cfBuilder.setBlockCacheEnabled(false); // 关闭数据块缓存
        cfBuilder.setCacheDataOnWrite(false);
        cfBuilder.setPrefetchBlocksOnOpen(false);

        // 版本配置
        cfBuilder.setMaxVersions(1);
        cfBuilder.setTimeToLive(7 * 24 * 3600); // 7天TTL，日志数据保存时间短

        // 大块大小提高压缩效率
        cfBuilder.setBlocksize(256 * 1024); // 256KB
        cfBuilder.setInMemory(false);

        // 优化写入性能
        cfBuilder.setScope(1); // 复制范围

        tableBuilder.setColumnFamily(cfBuilder.build());

        admin.createTable(tableBuilder.build());

        System.out.println("日志数据表创建成功：" + tableName.getNameAsString());
        printTableConfiguration("日志数据列族", cfBuilder.build());
    }

    /**
     * OLAP分析表优化配置
     * 特点：大范围扫描，聚合查询，需要快速读取
     */
    public static void createAnalyticsOptimizedTable(Admin admin) throws Exception {
        TableName tableName = TableName.valueOf("analytics_data");

        if (admin.tableExists(tableName)) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }

        TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(tableName);

        tableBuilder.setMaxFileSize(8L * 1024 * 1024 * 1024); // 8GB
        tableBuilder.setMemStoreFlushSize(1024L * 1024 * 1024); // 1GB，大内存刷新

        ColumnFamilyDescriptorBuilder cfBuilder = ColumnFamilyDescriptorBuilder
                .newBuilder("metrics".getBytes());

        // 平衡压缩比和读取性能
        cfBuilder.setCompressionType(Compression.Algorithm.LZ4);
        cfBuilder.setCompactionCompressionType(Compression.Algorithm.SNAPPY);

        // 数据块编码优化扫描性能
        cfBuilder.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);

        // 不使用BloomFilter，因为主要是大范围扫描
        cfBuilder.setBloomFilterType(BloomType.NONE);

        // 优化大范围扫描的缓存配置
        cfBuilder.setBlockCacheEnabled(true);
        cfBuilder.setCacheDataOnWrite(false);
        cfBuilder.setPrefetchBlocksOnOpen(true); // 预取有利于顺序扫描
        // 注意：索引缓存和L1缓存由HBase系统管理

        cfBuilder.setMaxVersions(1);
        //cfBuilder.setMinVersions(0);
        cfBuilder.setBlocksize(1024 * 1024); // 1MB大块，提高扫描效率
        cfBuilder.setInMemory(false);

        tableBuilder.setColumnFamily(cfBuilder.build());

        admin.createTable(tableBuilder.build());

        System.out.println("分析数据表创建成功：" + tableName.getNameAsString());
        printTableConfiguration("分析数据列族", cfBuilder.build());
    }

    /**
     * 生成时序数据的预分割键
     */
    private static byte[][] generateTimeSeriesSplitKeys() {
        String[] devicePrefixes = {"dev001", "dev002", "dev003", "dev004", "dev005"};
        byte[][] splitKeys = new byte[devicePrefixes.length - 1][];

        for (int i = 0; i < devicePrefixes.length - 1; i++) {
            splitKeys[i] = Bytes.toBytes(devicePrefixes[i + 1]);
        }

        return splitKeys;
    }

    /**
     * 打印表配置信息
     */
    private static void printTableConfiguration(String desc, ColumnFamilyDescriptor cf) {
        System.out.println("\n=== " + desc + " ===");
        System.out.println("压缩算法: " + cf.getCompressionType());
        System.out.println("Compaction压缩: " + cf.getCompactionCompressionType());
        System.out.println("数据块编码: " + cf.getDataBlockEncoding());
        System.out.println("BloomFilter类型: " + cf.getBloomFilterType());
        System.out.println("块缓存启用: " + cf.isBlockCacheEnabled());
        System.out.println("写时缓存数据: " + cf.isCacheDataOnWrite());
        System.out.println("预取块: " + cf.isPrefetchBlocksOnOpen());
        System.out.println("块大小: " + cf.getBlocksize() + " bytes");
        System.out.println("最大版本数: " + cf.getMaxVersions());
        System.out.println("TTL: " + cf.getTimeToLive() + " 秒");
        System.out.println("常驻内存: " + cf.isInMemory());
    }

    /**
     * 性能测试和监控方法
     */
    public static void performanceTest(Connection connection, String tableName) throws Exception {
        TableName table = TableName.valueOf(tableName);

        try (Table hTable = connection.getTable(table)) {
            long startTime = System.currentTimeMillis();

            // 写入测试数据
            for (int i = 0; i < 1000; i++) {
                Put put = new Put(Bytes.toBytes("row" + i));
                put.addColumn("cf".getBytes(), "col".getBytes(),
                        Bytes.toBytes("value" + i));
                hTable.put(put);
            }

            long writeTime = System.currentTimeMillis() - startTime;
            System.out.println("写入1000条记录耗时: " + writeTime + "ms");

            // 读取测试
            startTime = System.currentTimeMillis();
            Scan scan = new Scan();
            scan.setCacheBlocks(true);
            scan.setCaching(100);

            try (ResultScanner scanner = hTable.getScanner(scan)) {
                int count = 0;
                for (Result result : scanner) {
                    count++;
                }
                long scanTime = System.currentTimeMillis() - startTime;
                System.out.println("扫描" + count + "条记录耗时: " + scanTime + "ms");
            }
        }
    }

    /**
     * 获取表的统计信息
     */
    public static void getTableStats(Admin admin, String tableName) throws Exception {
        TableName table = TableName.valueOf(tableName);

        // 获取表的region信息
        try (RegionLocator locator = admin.getConnection().getRegionLocator(table)) {
            System.out.println("\n=== 表统计信息: " + tableName + " ===");
            System.out.println("Region数量: " + locator.getAllRegionLocations().size());

            locator.getAllRegionLocations().forEach(location -> {
                System.out.println("Region: " + location.getRegionInfo().getRegionNameAsString());
                System.out.println("  服务器: " + location.getServerName());
                System.out.println("  StartKey: " + Bytes.toString(location.getRegionInfo().getStartKey()));
                System.out.println("  EndKey: " + Bytes.toString(location.getRegionInfo().getEndKey()));
            });
        }
    }
}